package edu.vt.ece.rt.quorum.util;

import java.util.Collections;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;

import org.jgroups.Message;

import edu.vt.ece.rt.dstm.Context;
import edu.vt.ece.rt.dstm.ContextDelegator;
import edu.vt.ece.rt.dstm.QObject;
import edu.vt.ece.rt.dstm.TMProxy;
import edu.vt.ece.rt.quorum.ClustersManager;
import edu.vt.ece.rt.quorum.Resource;
import edu.vt.ece.rt.quorum.messsages.MessageHandler;
import edu.vt.ece.rt.quorum.messsages.QMessage;
import edu.vt.ece.rt.quorum.messsages.QMessage.MessageType;
import edu.vt.ece.rt.quorum.messsages.types.CommitConfirm;
import edu.vt.ece.rt.quorum.messsages.types.CommitRequest;
import edu.vt.ece.rt.util.Logger.Logger;

/*
 * This class is used to synchronise the message received in response of ReadRequest
 * @author Sudhanshu
 */
public class ReplyListStore {
	AtomicInteger counter;	
	public boolean requestFail=true;			//TO determine the request status
	/*
	 * Use priority as version object, QObject is already comparable
	 */
	Set<Context> contextList;			//FIXME: Might require change
	boolean waitOver;
	
	public ReplyListStore(int c){
		counter = new AtomicInteger(c);
		contextList = Collections.synchronizedSet(new HashSet<Context>());
		waitOver=false;
	}
	
	public void processMessage(Long txn, Set<Context> abortList, boolean commit) {
		Context context = ContextDelegator.getContext(txn);
		if(!commit){
			if(requestFail) requestFail=false;
			MessageHandler.commitRequestQueue.remove(txn);		//Remove yourself from queue
			context.setAborted();
			try {
				TMProxy.getTMProxy().getWriteChannel().send(new Message(null, new CommitConfirm(ContextDelegator.getContext(txn), false)));
				Resource.incrementMessages(TMProxy.getTMProxy().getWriteChannel().currentView.size());

//				Thread.sleep(1000L);
			} catch (Exception e) {
				e.printStackTrace();
			}			
			synchronized (this) {
				this.notifyAll();
			}
		} else {
			synchronized (contextList) {
				for(Context c: abortList){
					contextList.add(ContextDelegator.getContext(c.getContextId()));		//Use local copy to make hash work properly
				}
			}
			if(counter.decrementAndGet() == 0){			//if all messages are received as CommitAllowed, abort all other transactions
				if(requestFail) requestFail=false;
				
				synchronized (contextList) {
					Logger.debug("txn "+txn+" got commit so aborting conflicters");
					if (context.isRoot()) {
						for(Context c: contextList) {
							if (!CommitRequest.blast(context.getContextId(), c)) {
								Logger.debug("Aborint "+c.getContextId());
								ContextDelegator.getContext(c.getContextId()).setAborted();
							}	
							// TODO: Remove from all PR and PWs: already done in commit request
						}
					}
					
					//Send Confirm response to write Quorum
					
					try {
						TMProxy.getTMProxy().getWriteChannel().send(new Message(null, new CommitConfirm(ContextDelegator.getContext(txn), true)));
						Resource.incrementMessages(TMProxy.getTMProxy().getWriteChannel().currentView.size());
//						Thread.sleep(1000L);
					} catch (Exception e) {
						e.printStackTrace();
					}
					ContextDelegator.getContext(txn).setCommited();		//FIXME: Verify
				}
				
				synchronized (this) {
					this.notifyAll();					
				}
			}
		}
	}

}
